package com.by;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author ServivE
 * @version 0.1
 * @className Consumer
 * @description
 * @date 2024/10/19 16:32
 * @since jdk17
 */
public class ConsumerOne {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂  创建连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setVirtualHost("erp");//虚拟主机  默认是 /
        factory.setPort(5672);//连接端口号
        factory.setHost("192.168.203.130");//服务器地址 默认是localhost
        factory.setUsername("admin");
        factory.setPassword("123456");
        //2.创建连接
        Connection connection = factory.newConnection();
        //3.创建信道
        Channel channel = connection.createChannel();
        //4.声明队列
        String queueName = "work";
        /**
         *  参数说明:
         * 1.queue 队列的名称
         * 2.durable 队列是含持久化 true持久化 mq服务器重启任然存在 false 如果队列没有消息 队列会删除
         * 3.exclusive 是否独占 true独占的话 生产者消费者必须使用同一个连接
         * 4.autoDelete 队列是否自动测除
         * 5.arguments 附加参数
         */
        channel.queueDeclare(queueName, false, false, false, null);

        System.out.println("消费者one正在等待处理消息");
        //5.消费者消费消息
        /**
         * 参数说明:
         * 1.queue 臥列名称 当前消费者要从哪个队列取消息
         * 2.autoAck 消息应答模式 true自动应答
         * 3.deliverCallback 从mq服务器接收到消息的回调
         * 4.cancelcallback 取消回调
         */
        channel.basicConsume(queueName, true, (consumerTag, message) -> {
            System.out.println("consumerTag>>>" + consumerTag);
            String msg = new String(message.getBody());
            System.out.println("消费者one正在处理消息：" + msg);
            dowork(msg);
        }, consumerTag -> {
        });
        //注意 不用关闭释放资源  如果队列中有消息 会消费 ，没有会阻塞 等待消费
    }

    static void dowork(String msg) {
        byte[] tmp = msg.getBytes();
        for (int i = 0, len = tmp.length; i < len; i++) {
            try {
                //休眠1s
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}
